/*
* Copyright 2015 2dfire.com All right reserved. This software is the
* confidential and proprietary information of 2dfire.com ("Confidential
* Information"). You shall not disclose such Confidential Information and shall
* use it only in accordance with the terms of the license agreement you entered
* into with 2dfire.com.
*/
package com.dfire.mtm.taskNode;

import com.dfire.mtm.constant.TaskLife;
import com.dfire.mtm.executor.MtmRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author guipi
 * @description
 * @date 2018/10/23
 * ----------重大变更记录------------
 */
public abstract class AbstractTaskNode implements TaskNode {

    public static final String ROOT_NODE_NAME="root";

    protected final CopyOnWriteArrayList<TaskNode> parentNode = new CopyOnWriteArrayList<>();

    protected final CopyOnWriteArrayList<TaskNode> childrenNode = new CopyOnWriteArrayList<>();

    protected String name;

    protected ExecutorService executor;

    protected MtmRunnable mtmRunnable;

    protected final LinkedBlockingQueue<String> taskQueue = new LinkedBlockingQueue<>();

    /**
     * 按节点->任务->状态 存放任务状态
     */
    protected final ConcurrentHashMap<String,ConcurrentHashMap<String,TaskLife>> parentTaskStatus = new ConcurrentHashMap<>();

    /**
     * 按任务->状态 存放当前节点任务状态
     */
    protected final ConcurrentHashMap<String,TaskLife> taskStatus = new ConcurrentHashMap<>();

    /**
     * 按节点->任务->数据 存放数据
     */
    protected final ConcurrentHashMap<String,ConcurrentHashMap<String,BlockingQueue>> store = new ConcurrentHashMap<>();

    private static final Logger logger = LoggerFactory.getLogger(AbstractTaskNode.class);

    @Override
    public void setExecutor(ExecutorService executor){
        this.executor = executor;
    }

    public ExecutorService getExecutor(){
        return executor;
    }

    @Override
    public String getName() {
        return this.name;
    }

    @Override
    public List<TaskNode> getParentNode() {
        return this.parentNode;
    }

    @Override
    public void addParentNode(TaskNode taskNode) {
        this.parentNode.add(taskNode);
    }

    @Override
    public TaskNode getParentNode(String nodeName) {
        if(parentNode == null || parentNode.isEmpty()){
            return null;
        }
        return parentNode.stream().filter(taskNode -> taskNode.getName().equals(nodeName)).findFirst().orElse(null);
    }

    @Override
    public List<TaskNode> getChildrenNode() {
        return this.childrenNode;
    }

    @Override
    public void addChildrenNode(List<TaskNode> childrenNode) {
        if(CollectionUtils.isEmpty(childrenNode)){
            return;
        }
        this.childrenNode.addAll(childrenNode);
        childrenNode.forEach(taskNode -> taskNode.addParentNode(this));
    }

    @Override
    public TaskNode getChildrenNode(String nodeName) {
        if(childrenNode == null || childrenNode.isEmpty()){
            return null;
        }
        return childrenNode.stream().filter(taskNode -> taskNode.getName().equals(nodeName)).findFirst().orElse(null);
    }

    @Override
    public boolean notifyTaskStatus(String nodeName,String task, TaskLife taskLife) {
        ConcurrentHashMap<String,TaskLife> taskMap;
        if(parentTaskStatus.containsKey(nodeName)){
            taskMap = parentTaskStatus.get(nodeName);
        } else {
            taskMap = new ConcurrentHashMap<>();
            parentTaskStatus.put(nodeName,taskMap);
        }
        taskMap.put(task,taskLife);

        if(taskLife == TaskLife.ACTIVE){//激活时可以立即通知子节点
            taskStatus.put(task,taskLife);
            //通知下级任务状态
            if(!CollectionUtils.isEmpty(childrenNode)){
                childrenNode.forEach(taskNode -> {
                    taskNode.notifyTaskStatus(this.name,task,taskLife);
                });
            }
        }
        return true;
    }

    @Override
    public <T> boolean store(String task,String nodeName, T t) {
        if(!taskQueue.contains(task))taskQueue.add(task);

        if(!store.containsKey(nodeName)){
            BlockingQueue<T> queue = new LinkedBlockingQueue<>();
            queue.add(t);
            ConcurrentHashMap<String,BlockingQueue> taskMap= new ConcurrentHashMap<>();
            store.put(nodeName,taskMap);
            taskMap.put(task,queue);
            return true;
        }
        ConcurrentHashMap<String,BlockingQueue> taskMap = store.get(nodeName);
        BlockingQueue<T> queue;
        if(!taskMap.containsKey(task)){
            queue = new LinkedBlockingQueue<>();
            taskMap.put(task,queue);
        }else {
            queue = taskMap.get(task);
        }
        queue.add(t);
        return true;
    }

    @Override
    public void setRunnable(MtmRunnable mtmRunnable) {
        mtmRunnable.setTaskNode(this);
        this.mtmRunnable = mtmRunnable;
    }

    @Override
    public MtmRunnable getRunnable() {
        return this.mtmRunnable;
    }

    @Override
    public void execute() {
        while (true){
            if(executor.isShutdown()){
                return;
            }
            if(mtmRunnable==null){
                throw new NullPointerException("runnable is null");
            }
            if (mtmRunnable.isReady()) {
                executor.execute(mtmRunnable);
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public boolean existData() {
        return !taskQueue.isEmpty()&&!store.isEmpty();
    }

    @Override
    public <T> TaskData<T> nextData() {
        if(existData()){
            Iterator iterator = taskQueue.iterator();
            while (iterator.hasNext()){
                String task = (String) iterator.next();
                if(task!=null){
                    for(String node:store.keySet()){
                        T t =  getNextNodeData(task,store.get(node));
                        //队列里没数据时可以尝试清除task相关数据
                        if(t == null){
                            tryCleanTask(task);
                        } else {
                            return new TaskData<>(task,t);
                        }
                    }
                }
            }
        } else {
            for(ConcurrentHashMap<String,TaskLife> taskLifeMap : parentTaskStatus.values()){
                for(String task:taskLifeMap.keySet()){
                    tryCleanTask(task);
                }
            }
        }
        return null;
    }

    protected boolean tryCleanTask(String task){
        if(checkParentTaskIsFinished(task) && taskQueue.contains(task)){
            taskQueue.remove(task);
            for(String node :store.keySet()){
                ConcurrentHashMap<String,BlockingQueue> map = store.get(node);
                map.remove(task);
            }
            System.out.println(name +" "+ task +" is finished");
            //通知下级本节点此任务已完成
            if(!CollectionUtils.isEmpty(childrenNode)){
                childrenNode.forEach(taskNode -> {
                    taskNode.notifyTaskStatus(this.name,task,TaskLife.FINISH);
                });
            }
            return true;
        }
        return false;
    }

    /**
     * 检查父节点任务是否完成
     * @return
     * @param task
     */
    private boolean checkParentTaskIsFinished(String task){
        for(String node:parentTaskStatus.keySet()){
            ConcurrentHashMap<String,TaskLife> taskMap = parentTaskStatus.get(node);
            //任何一个父节点未完成时都不能结束
            if(taskMap.get(task)!=null && taskMap.get(task) == TaskLife.ACTIVE){
                return false;
            }
        }
        return true;
    }

    private <T> T getNextNodeData(String task, ConcurrentHashMap<String, BlockingQueue> map){
        if(map.containsKey(task)){
            if(!map.get(task).isEmpty()) return (T) map.get(task).poll();
        }
        return null;
    }

    @Override
    public <T> T getData(String task) {
        for(String node:store.keySet()){
            ConcurrentHashMap<String,BlockingQueue> map = store.get(node);
            BlockingQueue blockingQueue = map.get(task);
            if(blockingQueue !=null && !blockingQueue.isEmpty()){
                T t = (T) blockingQueue.poll();
                if(t!=null){
                    return t;
                }
            }
        }
        //无数据时尝试清除数据
        tryCleanTask(task);
        return null;
    }

    @Override
    public <T> boolean pushData(String task, T data) {
        List<TaskNode> childrenNodes = getChildrenNode();
        if (CollectionUtils.isEmpty(childrenNodes)) {
            logger.warn("childrenNodes is empty");
            return false;
        }
        for (TaskNode taskNode : childrenNodes) {
            taskNode.store(task, this.getName(), data);
        }
        return true;
    }

    @Override
    public <T> boolean pushData(String task, String nodeName, T data) {
        TaskNode taskNode = getChildrenNode(nodeName);
        if (taskNode == null) {
            logger.warn("childrenNodes is empty");
            return false;
        }
        taskNode.store(task, this.getName(), data);
        return true;
    }

    @Override
    public boolean taskIsFinished(String task) {
        TaskLife taskLife = taskStatus.get(task);
        if(taskLife == null){
            return false;
        }
        return taskLife == TaskLife.FINISH;
    }

    @Override
    public boolean stopTask(String task) {
        taskQueue.remove(task);
        for(String node :store.keySet()){
            ConcurrentHashMap<String,BlockingQueue> map = store.get(node);
            map.remove(task);
        }
        taskStatus.remove(task);
        for(String node:parentTaskStatus.keySet()){
            ConcurrentHashMap<String,TaskLife> taskMap = parentTaskStatus.get(node);
            taskMap.remove(task);
        }
        return true;
    }

    @Override
    public boolean stop() {
        executor.shutdown();
        taskStatus.clear();
        parentTaskStatus.clear();
        store.clear();
        taskQueue.clear();
        return true;
    }
}
